package com.boot;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.Consumer;

public class EventBus {
    private final ConcurrentHashMap<String, CopyOnWriteArraySet<Consumer<String>>> subscribes =
            new ConcurrentHashMap<>();

    //订阅事件
    public void subscribe(String topic, Consumer<String> subscriber) {
        subscribes.computeIfAbsent(topic, k -> new CopyOnWriteArraySet<>())
                .add(subscriber);
    }

    //取消订阅
    public void unsubscribe(String topic, Consumer<String> subscriber) {
        if (subscribes.containsKey(topic)) {
            subscribes.get(topic).remove(subscriber);
        }
    }

    //发布事件
    public void publish(String topic, String message) {
        if (subscribes.containsKey(topic)) {
            subscribes.get(topic).forEach(subscriber -> subscriber.accept(message));
        }
    }

    public static void main(String[] args) {
        EventBus eventBus = new EventBus();

        Consumer<String> subscriber1 = msg -> System.out.println("subscriber1 收到消息: " + msg);
        Consumer<String> subscriber2 = msg -> System.out.println("subscriber2 收到消息: " + msg);

        eventBus.subscribe("news", subscriber1);
        eventBus.subscribe("news", subscriber2);

        //多线程发送消息
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                eventBus.publish("news", "最新消息 from " + Thread.currentThread().getName());
            }).start();
        }
    }
}
